package com.algo.panelmn;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Method2CookieUVStep3 extends Configured implements Tool {
	public static class IdentityMapper extends org.apache.hadoop.mapreduce.Mapper<Text, Text, Text, Text> {
		public void map(Text key, Text value, Context context)
				throws IOException, InterruptedException {
			context.write(key, value);
		}
	}
	public static class Reducer extends org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, DoubleWritable>{
		public void reduce(Text key, Iterable<Text> values,Context context)
			 throws IOException, InterruptedException {
			double wt = 0;
			List<String> list = new ArrayList<String>();
			Iterator it = values.iterator();
			while(it.hasNext()){
				String lw = ((Text)it.next()).toString();
				String[] items = lw.split("\\^");
				if(items.length == 2)
					list.add(lw);
				else
					wt = wt + 1.0/Long.parseLong(lw);
			}
			for(String locpv : list){
				context.write(new Text(locpv), new DoubleWritable(wt));
			}
		}
	}
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        Job job = new Job(conf, "cookie-step3");
        job.setJarByClass(Method2CookieUVStep3.class);
        
        Path p1 = new Path(args[0]);
        Path p2 = new Path(args[1]);
        Path p3 = new Path(args[2]);
        MultipleInputs.addInputPath(job, p1, KeyValueTextInputFormat.class, IdentityMapper.class);
        MultipleInputs.addInputPath(job, p2, KeyValueTextInputFormat.class, IdentityMapper.class);
        FileOutputFormat.setOutputPath(job, p3);
        
        job.setReducerClass(Reducer.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
       
	if(RunChainJob2.isBatchCaids)   job.setNumReduceTasks(RunChainJob2.step3ReducerNum);
	else job.setNumReduceTasks(3); 
        return job.waitForCompletion(true) ? 1 : 0;
    }
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new Method2CookieUVStep3(), args);
        System.exit(res);
    }
}
